package system

import (
	"encoding/json"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/flipped-aurora/gin-vue-admin/server/global"
	"github.com/flipped-aurora/gin-vue-admin/server/model/system"
	"go.uber.org/zap"
	"sync"
)

// MQTTMessageManager MQTT消息管理器
type MQTTMessageManager struct {
	client      mqtt.Client
	subscribers map[uint64]chan *system.Site_message // 用户ID -> 消息通道
	mutex       sync.RWMutex
}

// NewMQTTMessageManager 创建新的MQTT消息管理器
func NewMQTTMessageManager(client mqtt.Client) *MQTTMessageManager {
	manager := &MQTTMessageManager{
		client:      client,
		subscribers: make(map[uint64]chan *system.Site_message),
		mutex:       sync.RWMutex{},
	}

	// 订阅站内信主题
	manager.subscribeToMessageTopic()

	return manager
}

// subscribeToMessageTopic 订阅站内信主题
func (m *MQTTMessageManager) subscribeToMessageTopic() {
	topic := "site_message/+"
	token := m.client.Subscribe(topic, 0, m.handleMessage)
	if token.Wait() && token.Error() != nil {
		global.GVA_LOG.Error("订阅站内信主题失败", zap.Error(token.Error()))
	} else {
		global.GVA_LOG.Info("已订阅站内信主题", zap.String("topic", topic))
	}
}

// handleMessage 处理接收到的MQTT消息
func (m *MQTTMessageManager) handleMessage(client mqtt.Client, msg mqtt.Message) {
	global.GVA_LOG.Info("接收到站内信MQTT消息",
		zap.String("topic", msg.Topic()),
		zap.String("payload", string(msg.Payload())))

	var siteMessage system.Site_message
	if err := json.Unmarshal(msg.Payload(), &siteMessage); err != nil {
		global.GVA_LOG.Error("解析站内信消息失败", zap.Error(err))
		return
	}

	// 发送消息给订阅的用户
	m.sendMessageToUser(&siteMessage)
}

// SubscribeUser 用户订阅消息
func (m *MQTTMessageManager) SubscribeUser(userID uint64) chan *system.Site_message {
	m.mutex.Lock()
	defer m.mutex.Unlock()

	// 创建用户的消息通道
	msgChan := make(chan *system.Site_message, 100)
	m.subscribers[userID] = msgChan

	global.GVA_LOG.Info("用户订阅站内信", zap.Uint64("user_id", userID))
	return msgChan
}

// UnsubscribeUser 用户取消订阅
func (m *MQTTMessageManager) UnsubscribeUser(userID uint64) {
	m.mutex.Lock()
	defer m.mutex.Unlock()

	if msgChan, exists := m.subscribers[userID]; exists {
		close(msgChan)
		delete(m.subscribers, userID)
		global.GVA_LOG.Info("用户取消订阅站内信", zap.Uint64("user_id", userID))
	}
}

// sendMessageToUser 发送消息给指定用户
func (m *MQTTMessageManager) sendMessageToUser(message *system.Site_message) {
	m.mutex.RLock()
	defer m.mutex.RUnlock()

	if message.User_id == nil {
		global.GVA_LOG.Warn("消息用户ID为空")
		return
	}

	userID := uint64(*message.User_id)
	if msgChan, exists := m.subscribers[userID]; exists {
		select {
		case msgChan <- message:
			global.GVA_LOG.Info("消息发送成功", zap.Uint64("user_id", userID))
		default:
			global.GVA_LOG.Warn("用户消息通道已满", zap.Uint64("user_id", userID))
		}
	} else {
		global.GVA_LOG.Debug("用户不在线", zap.Uint64("user_id", userID))
	}
}

// PublishMessage 发布站内信消息
func (m *MQTTMessageManager) PublishMessage(message *system.Site_message) error {
	if message.User_id == nil {
		return fmt.Errorf("消息用户ID为空")
	}

	// 序列化消息
	payload, err := json.Marshal(message)
	if err != nil {
		return fmt.Errorf("序列化消息失败: %w", err)
	}

	// 发布到用户特定的主题
	topic := fmt.Sprintf("site_message/%d", *message.User_id)
	token := m.client.Publish(topic, 0, false, payload)

	if token.Wait() && token.Error() != nil {
		return fmt.Errorf("发布消息失败: %w", token.Error())
	}

	global.GVA_LOG.Info("站内信消息发布成功",
		zap.String("topic", topic),
		zap.Uint64("user_id", uint64(*message.User_id)))

	return nil
}

// BroadcastMessage 广播消息给所有在线用户
func (m *MQTTMessageManager) BroadcastMessage(message *system.Site_message) error {
	// 序列化消息
	payload, err := json.Marshal(message)
	if err != nil {
		return fmt.Errorf("序列化消息失败: %w", err)
	}

	// 发布到广播主题
	topic := "site_message/broadcast"
	token := m.client.Publish(topic, 0, false, payload)

	if token.Wait() && token.Error() != nil {
		return fmt.Errorf("广播消息失败: %w", token.Error())
	}

	global.GVA_LOG.Info("站内信广播消息发布成功", zap.String("topic", topic))
	return nil
}

// GetOnlineUsers 获取在线用户数量
func (m *MQTTMessageManager) GetOnlineUsers() int {
	m.mutex.RLock()
	defer m.mutex.RUnlock()
	return len(m.subscribers)
}
